package com.boot.study.api

import org.apache.flink.api.common.functions.{FilterFunction, MapFunction, ReduceFunction, RichMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object TransformTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置全局并行度
    env.setParallelism(1)

    // 读取文件
    val inputSteam: DataStream[String] = env.readTextFile("D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt")
    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    //    // 分组聚合,输出每个传感器当前最小值
    //    val aggStream: DataStream[SensorReading] = dataStream
    //      .keyBy("id") // 根据id分组
    //      .minBy("temperature")
    //    aggStream.print()

    //    // 需要当前最小的温度值和最大的时间搓，使用reduce
    //    val reduceStream: DataStream[SensorReading] = dataStream
    //      .keyBy("id")
    //      .reduce((cur, news) => SensorReading(cur.id, news.timeStamp, cur.temperature.min(news.temperature)))
    //    reduceStream.print()

    // 降传感器分为高温流和低温流
    val splitStream: SplitStream[SensorReading] = dataStream.split(data => {
      if (data.temperature > 30.0) Seq("high") else Seq("low")
    })
    val highTempStream: DataStream[SensorReading] = splitStream.select("high")
    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")

    //    highTempStream.print("high")
    //    lowTempStream.print("low")

    //    // 合流 connect
    //    val warningStream: DataStream[(String, Double)] = highTempStream.map(data => (data.id, data.temperature))
    //    val connectStream: ConnectedStreams[(String, Double), SensorReading] = warningStream.connect(lowTempStream)
    //
    //    // 使用CoMap对数据进行分别处理
    //    val coMapResultStream: DataStream[Any] = connectStream.map(
    //      warningData => (warningData._1, warningData._2),
    //      lowTempData => (lowTempData.id, "healthy")
    //    )
    //    coMapResultStream.print("coMap")

    // union合流
    val unionStream: DataStream[SensorReading] = highTempStream.union(lowTempStream)
    unionStream.print()

    // 执行
    env.execute("transform test")
  }
}

// 自定义reduce函数
class MyReduceFunction extends ReduceFunction[SensorReading] {
  override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
    SensorReading(value1.id, value2.timeStamp, value1.temperature.min(value2.temperature))
  }
}

// 自定义函数类
class MyFilter extends FilterFunction[SensorReading] {
  override def filter(value: SensorReading): Boolean = value.id.startsWith("sensor_1")
}

class MyMapper extends MapFunction[SensorReading, String] {
  override def map(value: SensorReading): String = value.id + "temperature"
}

// 富函数，可以获取运行时上下文，以及一些生命周期
//class MyRichMapper extends RichMapFunction[SensorReading, String] {
//  override def open(parameters: Configuration): Unit = {
//    // 做一些初始化的操作，比如数据库的连接
//    // getRuntimeContext()
//  }
//
//  override def map(value: SensorReading): String = value.id + "temperature"
//
//  override def close(): Unit = {
//    // 一般收尾工作，比如关闭数据库连接，或者清空状态
//  }
//}
